/* 
 *  Copyright (c) 2011—2013 Panguso.com. All rights reserved. 
 *  File Name:  CassandraUtil.java                                      
 *  File Contents Outline : cassandra数据库的连接类
 *  File  Create Date:   2012年7月30日
 *  File Version#: Revision#: #1                  
 *  File  Latest Modify Date : 2013年1月16日
 *  File Author :  bachuan@panguso.com
 */
package storage.util;

import java.nio.ByteBuffer;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.List;
import java.util.logging.Level;
import java.util.logging.Logger;
import org.apache.cassandra.thrift.*;
import org.apache.thrift1.protocol.TBinaryProtocol;
import org.apache.thrift1.protocol.TProtocol;
import org.apache.thrift1.transport.TSocket;
import org.apache.thrift1.transport.TTransport;
import org.apache.thrift1.transport.TTransportException;
import org.apache.thrift1.transport.TFramedTransport;

import cn.edu.thu.log.web.service.PropertiesReaderServices;
import cn.edu.thu.log.web.service.impl.PropertiesReaderServicesImpl;

public class CassandraUtil {

    private static HashMap<String, TTransport> ttMap = new HashMap<String, TTransport>();
    private static HashMap<String, Cassandra.Client> clientMap = new HashMap<String, Cassandra.Client>();

    /**
     * 向链接库获得一个指定keyspace的数据库链接
     *
     * @param keySpace 指定的keyspace名称
     * @return 返回的数据库连接
     */
    public static Cassandra.Client getInstance(final String keySpace) {
        TTransport tt = ttMap.get(keySpace);
        Cassandra.Client client = clientMap.get(keySpace);
        PropertiesReaderServices x1 = PropertiesReaderServicesImpl.instance();
        if (tt == null) {
            try {
            //	System.out.println(x1.getProperties("dbConnection"));
                tt = new TFramedTransport(new TSocket(x1.getProperties("dbConnection"), 9160));
                TProtocol proto = new TBinaryProtocol(tt);
                client = new Cassandra.Client(proto);
                if(tt == null ) 
                	return null;
                tt.open();
                ttMap.put(keySpace, tt);
                clientMap.put(keySpace, client);
            } catch (TTransportException ex) {
                Logger.getLogger(CassandraUtil.class.getName()).log(Level.SEVERE, null, ex);
            }
        }

        return client;
    }

    public static void open(final String keySpace) {
        TTransport tt = ttMap.get(keySpace);
        if (tt != null) {
            if (!tt.isOpen()) {
                try {
                    tt.open();
                } catch (TTransportException ex) {
                    Logger.getLogger(CassandraUtil.class.getName()).log(Level.SEVERE, null, ex);
                }
            }
        }
    }

    /**
     * 关闭一个指定keyspace的数据库连接
     *
     * @param keySpace 指定的keyspace名称
     */
    public static void close(final String keySpace) {
        TTransport tt = ttMap.get(keySpace);
        if (tt != null) {
            if (tt.isOpen()) {
                tt.close();
            }
        }
    }

    private static SlicePredicate buildSlicePredicate(String start, String end) {
        
        SlicePredicate slicePredicate = new SlicePredicate();
        SliceRange sliceRange = new SliceRange();
        sliceRange.setStart(ByteBuffer.wrap(start.getBytes()));
        sliceRange.setFinish(ByteBuffer.wrap(end.getBytes()));
        slicePredicate.setSlice_range(sliceRange);

        return slicePredicate;
    }

    private static KeyRange buildKeyRange(String start, String end) {
        KeyRange keyrange = new KeyRange();
        keyrange.setStart_key(ByteBuffer.wrap(start.getBytes()));
        keyrange.setEnd_key(ByteBuffer.wrap(end.getBytes()));

        return keyrange;
    }

    private static IndexClause buildIndexClause(String column, String value, IndexOperator op, String startKey) {
        IndexClause indexClause = new IndexClause();
        IndexExpression ie = new IndexExpression();
        ie.setColumn_name(ByteBuffer.wrap(column.getBytes()));
        ie.setOp(op);
        ie.setValue(ByteBuffer.wrap(value.getBytes()));
        indexClause.addToExpressions(ie);
        indexClause.setStart_key(ByteBuffer.wrap(startKey.getBytes()));
       // indexClause.setCount(limitrow);

        return indexClause;
    }

    private static IndexClause buildIndexClause(String column1, String value1, String column2, String value2, IndexOperator op, String startKey) {
        IndexClause indexClause = new IndexClause();
        IndexExpression ie = new IndexExpression();
        ie.setColumn_name(ByteBuffer.wrap(column1.getBytes()));
        ie.setOp(op);
        ie.setValue(ByteBuffer.wrap(value1.getBytes()));
        indexClause.addToExpressions(ie);
        IndexExpression ie1 = new IndexExpression();
        ie1.setColumn_name(ByteBuffer.wrap(column2.getBytes()));
        ie1.setOp(op);
        ie1.setValue(ByteBuffer.wrap(value2.getBytes()));
        indexClause.addToExpressions(ie1);
        indexClause.setStart_key(ByteBuffer.wrap(startKey.getBytes()));
        return indexClause;
    }

    private static Column buildColumn(String columnName, String value) {
        Column column = new Column();
        column.setName(columnName.getBytes());
        column.setValue(value.getBytes());
        column.setTimestamp(System.currentTimeMillis());

        return column;
    }

    private static ColumnPath buildColumnPath(String columnFamily) {
        ColumnPath columnPath = new ColumnPath();
        columnPath.setColumn_family(columnFamily);
        return columnPath;
    }

    /**
     * 从ColumnFamily中选取所有行的Rowkey
     *
     * @param keySpace keySpace名
     * @param columnFamily 列族名
     * @return Cassandra结果集，可以用next()方法遍历，用getKey()方法获得Rowkey
     * @throws SQLException 若SQL语句不合法
     * @throws ClassNotFoundException 若找不到驱动类
     */
    public static List<KeySlice> selectAll(final String keySpace, final String columnFamily) throws Exception {
        List<KeySlice> result;
        Cassandra.Client client = getInstance(keySpace);

        client.set_keyspace(keySpace);
        ColumnParent columnParent = new ColumnParent(columnFamily);

        SlicePredicate slicePredicate = buildSlicePredicate("", "");
        KeyRange keyrange = buildKeyRange("", "");

        result = client.get_range_slices(columnParent, slicePredicate, keyrange, ConsistencyLevel.ONE);

        return result;
    }

    /**
     * 从ColumnFamily中选取所有行的Rowkey
     *
     * @param keySpace keySpace名
     * @param columnFamily 列族名
     * @return Cassandra结果集，可以用next()方法遍历，用getKey()方法获得Rowkey
     * @throws SQLException 若SQL语句不合法
     * @throws ClassNotFoundException 若找不到驱动类
     */
    public static List<KeySlice> selectAllInOneColumn(final String keySpace, final String columnFamily, String columnName) throws Exception {
        List<KeySlice> result;
        Cassandra.Client client = getInstance(keySpace);

        client.set_keyspace(keySpace);
        ColumnParent columnParent = new ColumnParent(columnFamily);

        SlicePredicate slicePredicate = buildSlicePredicate(columnName, columnName);
        KeyRange keyrange = buildKeyRange("", "");

        result = client.get_range_slices(columnParent, slicePredicate, keyrange, ConsistencyLevel.ONE);

        return result;
    }
 /**
     * 从ColumnFamily中选取所有行的Rowkey
     *
     * @param keySpace keySpace名
     * @param columnFamily 列族名
     * @return Cassandra结果集，可以用next()方法遍历，用getKey()方法获得Rowkey
     * @throws SQLException 若SQL语句不合法
     * @throws ClassNotFoundException 若找不到驱动类
     */
    public static List<KeySlice> selectAllInOneColumnByAnother(final String keySpace, final String columnFamily, String columnName1, String value, String columnName2) throws Exception {
        List<KeySlice> result;
        Cassandra.Client client = getInstance(keySpace);

        client.set_keyspace(keySpace);
        ColumnParent columnParent = new ColumnParent(columnFamily);

        if (columnName1.toLowerCase().equals("key")) {
            SlicePredicate slicePredicate = buildSlicePredicate("", "");
            KeyRange keyrange = buildKeyRange(value, value);

            result = client.get_range_slices(columnParent, slicePredicate, keyrange, ConsistencyLevel.ONE);
        } else {
          IndexClause indexClause = buildIndexClause(columnName1, value, IndexOperator.EQ, "");
            SlicePredicate slicePredicate = buildSlicePredicate(columnName2, columnName2);

            result = client.get_indexed_slices(columnParent, indexClause, slicePredicate, ConsistencyLevel.ONE);
        }
        return result;
    }

    /**
     * 从ColumnFamily中选取指定列的内容为指定值的Rowkey
     *
     * @param keySpace keySpace名
     * @param columnFamily 列族名
     * @param column 指定列的列名
     * @param value 指定列的值
     * @return Cassandra结果集，可以用next()方法遍历，用getKey()方法获得Rowkey
     * @throws SQLException
     */
    public static List<KeySlice> select(final String keySpace, final String columnFamily, final String column, final String value) throws Exception {
        List<KeySlice> result;
        Cassandra.Client client = getInstance(keySpace);

        client.set_keyspace(keySpace);
        ColumnParent columnParent = new ColumnParent(columnFamily);

        if (column.toLowerCase().equals("key")) {
            SlicePredicate slicePredicate = buildSlicePredicate("", "");
            KeyRange keyrange = buildKeyRange(value, value);

            result = client.get_range_slices(columnParent, slicePredicate, keyrange, ConsistencyLevel.ONE);
        } else {
            IndexClause indexClause = buildIndexClause(column, value, IndexOperator.EQ, "");
            SlicePredicate slicePredicate = buildSlicePredicate("", "");

            result = client.get_indexed_slices(columnParent, indexClause, slicePredicate, ConsistencyLevel.ONE);
        }
        return result;
    } /**
     * 从ColumnFamily中选取指定列的内容为指定值的Rowkey
     *
     * @param keySpace keySpace名
     * @param columnFamily 列族名
     * @param column 指定列的列名
     * @param value 指定列的值
     * @return Cassandra结果集，可以用next()方法遍历，用getKey()方法获得Rowkey
     * @throws SQLException
     */
    public static List<KeySlice> selectByTwoColumn(final String keySpace, final String columnFamily, final String column, final String value,final String column1, final String value1) throws Exception {
        List<KeySlice> result;
        Cassandra.Client client = getInstance(keySpace);

        client.set_keyspace(keySpace);
        ColumnParent columnParent = new ColumnParent(columnFamily);

        if (column.toLowerCase().equals("key")) {
            SlicePredicate slicePredicate = buildSlicePredicate("", "");
            KeyRange keyrange = buildKeyRange(value, value);

            result = client.get_range_slices(columnParent, slicePredicate, keyrange, ConsistencyLevel.ONE);
        } else {
            IndexClause indexClause = buildIndexClause(column, value,column1, value1, IndexOperator.EQ, "");
            SlicePredicate slicePredicate = buildSlicePredicate("", "");

            result = client.get_indexed_slices(columnParent, indexClause, slicePredicate, ConsistencyLevel.ONE);
        }
        return result;
    }

    /**
     * 更新数据库中某行某列的信息
     *
     * @param keySpace keySpace名
     * @param columnFamily 列族名
     * @param rowKey 对应行的行键
     * @param column 目标列的列名
     * @param value 目标列的值
     * @throws SQLException
     */
    public static void update(final String keySpace, final String columnFamily, final String rowKey, final String columnName, final String value) throws Exception {
        Cassandra.Client client = getInstance(keySpace);

        client.set_keyspace(keySpace);
        ColumnParent columnParent = new ColumnParent(columnFamily);

        Column column = buildColumn(columnName, value);

        client.insert(ByteBuffer.wrap(rowKey.getBytes()), columnParent, column, ConsistencyLevel.ONE);
    }

    public static void delete(final String keySpace, final String columnFamily, final String rowKey, final String columnName, final String value) throws Exception {
        Cassandra.Client client = getInstance(keySpace);

        client.set_keyspace(keySpace);
        ColumnParent columnParent = new ColumnParent(columnFamily);

        ColumnPath columnPath = buildColumnPath(columnFamily);
        Column column = buildColumn(columnName, value);
        client.remove(ByteBuffer.wrap(rowKey.getBytes()), columnPath, System.currentTimeMillis(), ConsistencyLevel.ONE);
//          List<KeySlice> result = select(keySpace, columnFamily, columnName, value);
//          for (KeySlice ks : result) {
//            client.remove(ByteBuffer.wrap(rowKey.getBytes())), columnPath, System.currentTimeMillis(), ConsistencyLevel.ALL);
//        }
    }

    /**
     * 删除指定column列的值为value的行的信息
     *
     * @param keySpace keySpace名
     * @param columnFamily 列族名
     * @param column 指定列的列名
     * @param value 指定列的value名
     * @throws SQLException
     */
    public static void delete(final String keySpace, final String columnFamily, final String column, final String value) throws Exception {
        Cassandra.Client client = getInstance(keySpace);

        client.set_keyspace(keySpace);
        //ColumnParent columnParent = new ColumnParent(columnFamily);

        ColumnPath columnPath = buildColumnPath(columnFamily);
        List<KeySlice> result = select(keySpace, columnFamily, column, value);
        for (KeySlice ks : result) {
            client.remove(ByteBuffer.wrap(ks.getKey()), columnPath, System.currentTimeMillis(), ConsistencyLevel.ONE);
        }
    }
}
